package com.etc;

import com.google.gson.Gson;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import java.io.IOException;
import java.text.SimpleDateFormat;


/**
 * 日志清洗
 * 1.通过map采集hdfs的今天的数据
 * 2.得到每一行的数据格式是json
 * 3.通过gson技术 进行json字符串和对象的转换
 * 4.判断对象中每个字段是否是我们需要的字段
 * 5.添加userid字段进去
 * 6.发送出去
 */
public class DataMapper extends Mapper<LongWritable, Text,Text , NullWritable> {

    Text k = null;
    NullWritable v = null;
    SimpleDateFormat sdf = null;
    MultipleOutputs<Text,NullWritable> mos = null;  //多路输出器


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        k = new Text();
        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        mos = new MultipleOutputs<Text, NullWritable>(context);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();


        Gson gson = new Gson();
        //将json字符串转为对象
        LogDataBean.HeaderBean headerBean = gson.fromJson(line,LogDataBean.class).getHeader();

        if (headerBean.getSdk_ver().equals("")){
            return;
        }

        if (headerBean.getTime_zone().equals("")){
            return;
        }
        if (headerBean.getCommit_id().equals("")){
            return;
        }
        if (headerBean.getCommit_time().equals("")){
            return;
        }
        if (headerBean.getPid().equals("")){
            return;
        }
        if (headerBean.getApp_token().equals("")){
            return;
        }
        if (headerBean.getApp_id().equals("")){
            return;
        }
        if (headerBean.getDevice_id().equals("")){
            return;
        }
        if (headerBean.getDevice_id_type().equals("")){
            return;
        }
        if (headerBean.getRelease_channel().equals("")){
            return;
        }
        if (headerBean.getApp_ver_name().equals("")){
            return;
        }
        if (headerBean.getApp_ver_code().equals("")){
            return;
        }
        if (headerBean.getOs_name().equals("")){
            return;
        }
        if (headerBean.getOs_ver().equals("")){
            return;
        }
        if (headerBean.getLanguage().equals("")){
            return;
        }
        if (headerBean.getCountry().equals("")){
            return;
        }
        if (headerBean.getManufacture().equals("")){
            return;
        }
        if (headerBean.getDevice_model().equals("")){
            return;
        }
        if (headerBean.getResolution().equals("")){
            return;
        }
        if (headerBean.getNet_type().equals("")){
            return;
        }



        String user_id = "";
        if (headerBean.getOs_name().equals("android")){
            user_id = headerBean.getAndroid_id();
        }else {
            user_id = headerBean.getDevice_id();
        }


        /**
         * 输出结果
         */

        headerBean.setUser_id(user_id);
        k.set(headerBean.toString());
        if("android".equals(headerBean.getOs_name())){
            mos.write(k, v, "android/android");
        }else{
            mos.write(k, v, "ios/ios");
        }



    }


    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        mos.close();
    }

}
